package cn.tank.data.sync.kafka;

import cn.tank.data.sync.core.MessageSendService;
import cn.tank.data.sync.core.common.Message;
import cn.tank.data.sync.kafka.common.KafkaManager;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.Future;

/************************************************
 * @Description:kafka消息发布实现
 * @Author: tancyong
 * @Version: 1.0
 ************************************************/
public class KafkaMessagePublisher implements MessageSendService {

    private static final Logger log = LoggerFactory.getLogger(KafkaMessagePublisher.class);

    private static final String DATA_TYPE_HEADER = "DATATYPE";

    private final KafkaManager kafkaManager;

    public KafkaMessagePublisher(KafkaManager kafkaManager) {
        this.kafkaManager = kafkaManager;
    }

    /**
     * 使用kafka的方式实现时 channel为消息主题 即toptic
     * @param channel
     * @param messge
     */
    @Override
    public void sendMessage(String channel, Message messge) {
        try {

            ProducerRecord<String, String> record = new ProducerRecord<>(channel, messge.getData() == null ? "" : JSONObject.toJSONString(messge.getData()));
            //kafka的实现，消息事件类型放入到消息头中
            record.headers().add(new RecordHeader(DATA_TYPE_HEADER, messge.getDataType().getBytes(StandardCharsets.UTF_8)));

            Future<RecordMetadata> ft = kafkaManager.getProducer().send(record);

            RecordMetadata meta = ft.get();
            log.info("kafka send message metadata：{}", meta);
        } catch (Exception e) {
            log.error("kafka send message error", e);
            throw new RuntimeException(e);
        }

    }
}
